--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit 91a0879912b5451a86cfcc04478d6dd7d1825c7e
Parents : 1ed19c3
Author : Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-04-13T20:55:32-05:00
feat(stickers): implement sticker management features including creation, deletion, listing, and import/export functionality, along with database schema updates for user stickers
Changes
9 files changed, 655 insertions(+), 10 deletions(-)
Diff
diff --git a/meshchatx/meshchat.py b/meshchatx/meshchat.py
index 3f47c34d..abd4d6ba 100644
--- a/meshchatx/meshchat.py
+++ b/meshchatx/meshchat.py
@@ -25,6 +25,7 @@ import time
import traceback
import webbrowser
import io
+import subprocess
import zipfile
import fnmatch
from datetime import UTC, datetime, timedelta
@@ -103,6 +104,12 @@ from meshchatx.src.backend.persistent_log_handler import PersistentLogHandler
from meshchatx.src.backend.recovery import CrashRecovery, HealthMonitor
from meshchatx.src.backend.rnprobe_handler import RNProbeHandler
from meshchatx.src.backend.sideband_commands import SidebandCommands
+from meshchatx.src.backend.sticker_utils import (
+ build_export_document,
+ mime_for_image_type,
+ sanitize_sticker_name,
+ validate_export_document,
+)
from meshchatx.src.backend.telemetry_utils import Telemeter
from meshchatx.src.backend.web_audio_bridge import WebAudioBridge
from meshchatx.src.version import __version__ as app_version
@@ -4831,6 +4838,12 @@ class ReticulumMeshChat:
self.database.misc.delete_all_user_icons()
return web.json_response({"message": "All LXMF icons cleared"})
+ @routes.delete("/api/v1/maintenance/stickers")
+ async def maintenance_clear_stickers(request):
+ identity_hash = self.identity.hash.hex()
+ n = self.database.stickers.delete_all_for_identity(identity_hash)
+ return web.json_response({"message": "Stickers cleared", "deleted": n})
+
# maintenance - export messages
@routes.get("/api/v1/maintenance/messages/export")
async def maintenance_export_messages(request):
@@ -9170,6 +9183,110 @@ class ReticulumMeshChat:
self.database.map_drawings.update_drawing(drawing_id, name, drawing_data)
return web.json_response({"message": "Drawing updated successfully"})
+ @routes.get("/api/v1/stickers")
+ async def stickers_list(request):
+ identity_hash = self.identity.hash.hex()
+ rows = self.database.stickers.list_for_identity(identity_hash)
+ return web.json_response({"stickers": [dict(r) for r in rows]})
+
+ @routes.post("/api/v1/stickers")
+ async def stickers_create(request):
+ identity_hash = self.identity.hash.hex()
+ try:
+ data = await request.json()
+ except (json.JSONDecodeError, ValueError):
+ return web.json_response({"error": "invalid_json"}, status=400)
+ image_b64 = data.get("image_bytes")
+ if not isinstance(image_b64, str) or not image_b64.strip():
+ return web.json_response({"error": "missing_image_bytes"}, status=400)
+ try:
+ raw = base64.b64decode(image_b64.strip(), validate=True)
+ except (ValueError, TypeError):
+ return web.json_response({"error": "invalid_base64"}, status=400)
+ name = sanitize_sticker_name(data.get("name"))
+ image_type = data.get("image_type")
+ src = data.get("source_message_hash")
+ src = src if isinstance(src, str) else None
+ try:
+ row = self.database.stickers.insert(
+ identity_hash,
+ name,
+ image_type,
+ raw,
+ src,
+ )
+ except ValueError as e:
+ return web.json_response({"error": str(e)}, status=400)
+ if row is None:
+ return web.json_response({"error": "duplicate_sticker"}, status=409)
+ return web.json_response({"sticker": row})
+
+ @routes.delete("/api/v1/stickers/{sticker_id}")
+ async def stickers_delete(request):
+ identity_hash = self.identity.hash.hex()
+ sticker_id = int(request.match_info.get("sticker_id", "0"))
+ ok = self.database.stickers.delete(sticker_id, identity_hash)
+ if not ok:
+ return web.json_response({"error": "not_found"}, status=404)
+ return web.json_response({"message": "deleted"})
+
+ @routes.patch("/api/v1/stickers/{sticker_id}")
+ async def stickers_patch(request):
+ identity_hash = self.identity.hash.hex()
+ sticker_id = int(request.match_info.get("sticker_id", "0"))
+ try:
+ data = await request.json()
+ except (json.JSONDecodeError, ValueError):
+ return web.json_response({"error": "invalid_json"}, status=400)
+ if "name" not in data:
+ return web.json_response({"error": "missing_name"}, status=400)
+ name = sanitize_sticker_name(data.get("name"))
+ ok = self.database.stickers.update_name(sticker_id, identity_hash, name)
+ if not ok:
+ return web.json_response({"error": "not_found"}, status=404)
+ return web.json_response({"message": "updated"})
+
+ @routes.get("/api/v1/stickers/{sticker_id}/image")
+ async def stickers_get_image(request):
+ identity_hash = self.identity.hash.hex()
+ sticker_id = int(request.match_info.get("sticker_id", "0"))
+ row = self.database.stickers.get_row(sticker_id, identity_hash)
+ if row is None:
+ return web.json_response({"error": "not_found"}, status=404)
+ ct = mime_for_image_type(row["image_type"])
+ return web.Response(body=row["image_blob"], content_type=ct)
+
+ @routes.get("/api/v1/stickers/export")
+ async def stickers_export(request):
+ identity_hash = self.identity.hash.hex()
+ payloads = self.database.stickers.export_payloads_for_identity(
+ identity_hash
+ )
+ doc = build_export_document(
+ payloads,
+ datetime.now(UTC).isoformat(),
+ )
+ return web.json_response(doc)
+
+ @routes.post("/api/v1/stickers/import")
+ async def stickers_import(request):
+ identity_hash = self.identity.hash.hex()
+ try:
+ data = await request.json()
+ except (json.JSONDecodeError, ValueError):
+ return web.json_response({"error": "invalid_json"}, status=400)
+ replace = bool(data.get("replace_duplicates", False))
+ try:
+ items = validate_export_document(data)
+ except ValueError as e:
+ return web.json_response({"error": str(e)}, status=400)
+ result = self.database.stickers.import_payloads(
+ identity_hash,
+ items,
+ replace_duplicates=replace,
+ )
+ return web.json_response(result)
+
# get latest telemetry for all peers
@routes.get("/api/v1/telemetry/peers")
async def get_all_latest_telemetry(request):
@@ -10264,6 +10381,11 @@ class ReticulumMeshChat:
data["message_failed_bubble_color"]
)
+ if "message_waiting_bubble_color" in data:
+ self.config.message_waiting_bubble_color.set(
+ data["message_waiting_bubble_color"]
+ )
+
# update desktop settings
if "desktop_open_calls_in_separate_window" in data:
self.config.desktop_open_calls_in_separate_window.set(
@@ -11429,6 +11551,7 @@ class ReticulumMeshChat:
"message_outbound_bubble_color": ctx.config.message_outbound_bubble_color.get(),
"message_inbound_bubble_color": ctx.config.message_inbound_bubble_color.get(),
"message_failed_bubble_color": ctx.config.message_failed_bubble_color.get(),
+ "message_waiting_bubble_color": ctx.config.message_waiting_bubble_color.get(),
"translator_enabled": ctx.config.translator_enabled.get(),
"libretranslate_url": ctx.config.libretranslate_url.get(),
"desktop_open_calls_in_separate_window": ctx.config.desktop_open_calls_in_separate_window.get(),
@@ -11778,6 +11901,47 @@ class ReticulumMeshChat:
except Exception:
return False
+ def _convert_webm_opus_to_ogg(self, audio_bytes: bytes) -> bytes:
+ """Convert WebM/Opus audio to OGG/Opus using ffmpeg.
+
+ Browser MediaRecorder outputs Opus in a WebM container, but LXMF
+ AM_OPUS_OGG expects an OGG container. If ffmpeg is unavailable or
+ the input is already OGG, the original bytes are returned as-is.
+ """
+ if audio_bytes[:4] == b"OggS":
+ return audio_bytes
+
+ ffmpeg_path = shutil.which("ffmpeg")
+ if ffmpeg_path is None:
+ return audio_bytes
+
+ try:
+ result = subprocess.run( # noqa: S603
+ [
+ ffmpeg_path,
+ "-i",
+ "pipe:0",
+ "-c:a",
+ "libopus",
+ "-b:a",
+ "24k",
+ "-vbr",
+ "on",
+ "-f",
+ "ogg",
+ "pipe:1",
+ ],
+ input=audio_bytes,
+ capture_output=True,
+ timeout=30,
+ )
+ if result.returncode == 0 and len(result.stdout) > 0:
+ return result.stdout
+ except Exception as e:
+ print(f"WebM to OGG conversion failed: {e}")
+
+ return audio_bytes
+
# check if a destination is blocked
def is_destination_blocked(self, destination_hash: str, context=None) -> bool:
ctx = context or self.current_context
@@ -12444,6 +12608,11 @@ class ReticulumMeshChat:
lxmf_message.fields = {}
+ lxmf_message.fields[LXMF.FIELD_RENDERER] = LXMF.RENDERER_MARKDOWN
+
+ if self._is_contact(destination_hash, context=ctx):
+ lxmf_message.include_ticket = True
+
# add file attachments field
if file_attachments_field is not None:
# create array of [[file_name, file_bytes], [file_name, file_bytes], ...]
@@ -12464,9 +12633,12 @@ class ReticulumMeshChat:
# add audio field
if audio_field is not None:
+ audio_bytes = audio_field.audio_bytes
+ if audio_field.audio_mode == LXMF.AM_OPUS_OGG:
+ audio_bytes = self._convert_webm_opus_to_ogg(audio_bytes)
lxmf_message.fields[LXMF.FIELD_AUDIO] = [
audio_field.audio_mode,
- audio_field.audio_bytes,
+ audio_bytes,
]
# add telemetry field
diff --git a/meshchatx/src/backend/async_utils.py b/meshchatx/src/backend/async_utils.py
index 4625e4dc..e33ad3b9 100644
--- a/meshchatx/src/backend/async_utils.py
+++ b/meshchatx/src/backend/async_utils.py
@@ -62,11 +62,15 @@ class AsyncUtils:
"""
if AsyncUtils.main_loop and AsyncUtils.main_loop.is_running():
future = asyncio.run_coroutine_threadsafe(
- coroutine, AsyncUtils.main_loop,
+ coroutine,
+ AsyncUtils.main_loop,
)
with AsyncUtils._futures_lock:
AsyncUtils._pending_futures.append(future)
- if len(AsyncUtils._pending_futures) >= AsyncUtils._FUTURES_SWEEP_THRESHOLD:
+ if (
+ len(AsyncUtils._pending_futures)
+ >= AsyncUtils._FUTURES_SWEEP_THRESHOLD
+ ):
AsyncUtils._pending_futures = [
f for f in AsyncUtils._pending_futures if not f.done()
]
diff --git a/meshchatx/src/backend/config_manager.py b/meshchatx/src/backend/config_manager.py
index b301210b..6385a3a1 100644
--- a/meshchatx/src/backend/config_manager.py
+++ b/meshchatx/src/backend/config_manager.py
@@ -317,6 +317,11 @@ class ConfigManager:
"message_failed_bubble_color",
"#ef4444",
)
+ self.message_waiting_bubble_color = self.StringConfig(
+ self,
+ "message_waiting_bubble_color",
+ "#e5e7eb",
+ )
# announce caps: max rows stored per aspect (oldest dropped). Default 1000.
self.announce_max_stored_lxmf_delivery = self.IntConfig(
diff --git a/meshchatx/src/backend/database/__init__.py b/meshchatx/src/backend/database/__init__.py
index aee0668f..2b1f5744 100644
--- a/meshchatx/src/backend/database/__init__.py
+++ b/meshchatx/src/backend/database/__init__.py
@@ -18,6 +18,7 @@ from .misc import MiscDAO
from .provider import DatabaseProvider
from .ringtones import RingtoneDAO
from .schema import DatabaseSchema
+from .stickers import UserStickersDAO
from .telemetry import TelemetryDAO
from .telephone import TelephoneDAO
from .voicemails import VoicemailDAO
@@ -42,6 +43,7 @@ class Database:
self.ringtones = RingtoneDAO(self.provider)
self.contacts = ContactsDAO(self.provider)
self.map_drawings = MapDrawingsDAO(self.provider)
+ self.stickers = UserStickersDAO(self.provider)
self.debug_logs = DebugLogsDAO(self.provider)
self.access_attempts = AccessAttemptsDAO(self.provider)
self.crash_history = CrashHistoryDAO(self.provider)
diff --git a/meshchatx/src/backend/database/messages.py b/meshchatx/src/backend/database/messages.py
index 792eb4bb..00d2689e 100644
--- a/meshchatx/src/backend/database/messages.py
+++ b/meshchatx/src/backend/database/messages.py
@@ -59,9 +59,17 @@ class MessageDAO:
self.provider.execute(query, params)
- def update_lxmf_message_state(self, message_hash, state, progress,
- delivery_attempts, next_delivery_attempt_at,
- rssi=None, snr=None, quality=None):
+ def update_lxmf_message_state(
+ self,
+ message_hash,
+ state,
+ progress,
+ delivery_attempts,
+ next_delivery_attempt_at,
+ rssi=None,
+ snr=None,
+ quality=None,
+ ):
"""Lightweight update for delivery-state changes only.
Avoids re-serializing the full message (including base64 attachment
@@ -73,8 +81,17 @@ class MessageDAO:
"delivery_attempts = ?, next_delivery_attempt_at = ?, "
"rssi = ?, snr = ?, quality = ?, updated_at = ? "
"WHERE hash = ?",
- (state, progress, delivery_attempts, next_delivery_attempt_at,
- rssi, snr, quality, now, message_hash),
+ (
+ state,
+ progress,
+ delivery_attempts,
+ next_delivery_attempt_at,
+ rssi,
+ snr,
+ quality,
+ now,
+ message_hash,
+ ),
)
def get_lxmf_message_by_hash(self, message_hash):
diff --git a/meshchatx/src/backend/database/schema.py b/meshchatx/src/backend/database/schema.py
index b79cc352..28cf8bc8 100644
--- a/meshchatx/src/backend/database/schema.py
+++ b/meshchatx/src/backend/database/schema.py
@@ -13,7 +13,7 @@ def _validate_identifier(name: str, label: str = "identifier") -> str:
class DatabaseSchema:
- LATEST_VERSION = 43
+ LATEST_VERSION = 44
def __init__(self, provider: DatabaseProvider):
self.provider = provider
@@ -417,6 +417,20 @@ class DatabaseSchema:
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""",
+ "user_stickers": """
+ CREATE TABLE IF NOT EXISTS user_stickers (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ identity_hash TEXT NOT NULL,
+ name TEXT,
+ image_type TEXT NOT NULL,
+ image_blob BLOB NOT NULL,
+ content_hash TEXT NOT NULL,
+ source_message_hash TEXT,
+ created_at REAL NOT NULL,
+ updated_at REAL NOT NULL,
+ UNIQUE(identity_hash, content_hash)
+ )
+ """,
"lxmf_last_sent_icon_hashes": """
CREATE TABLE IF NOT EXISTS lxmf_last_sent_icon_hashes (
destination_hash TEXT PRIMARY KEY,
@@ -1119,6 +1133,28 @@ class DatabaseSchema:
"CREATE INDEX IF NOT EXISTS idx_lxmf_conversation_pins_pinned_at ON lxmf_conversation_pins(pinned_at)",
)
+ if current_version < 44:
+ self._safe_execute("""
+ CREATE TABLE IF NOT EXISTS user_stickers (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ identity_hash TEXT NOT NULL,
+ name TEXT,
+ image_type TEXT NOT NULL,
+ image_blob BLOB NOT NULL,
+ content_hash TEXT NOT NULL,
+ source_message_hash TEXT,
+ created_at REAL NOT NULL,
+ updated_at REAL NOT NULL,
+ UNIQUE(identity_hash, content_hash)
+ )
+ """)
+ self._safe_execute(
+ "CREATE INDEX IF NOT EXISTS idx_user_stickers_identity ON user_stickers(identity_hash)",
+ )
+ self._safe_execute(
+ "CREATE INDEX IF NOT EXISTS idx_user_stickers_identity_updated ON user_stickers(identity_hash, updated_at)",
+ )
+
# Update version in config
self._safe_execute(
"""
diff --git a/meshchatx/src/backend/database/stickers.py b/meshchatx/src/backend/database/stickers.py
new file mode 100644
index 00000000..ee5744e3
--- /dev/null
+++ b/meshchatx/src/backend/database/stickers.py
@@ -0,0 +1,229 @@
+import base64
+import sqlite3
+import time
+
+from meshchatx.src.backend import sticker_utils
+
+
+class UserStickersDAO:
+ def __init__(self, provider):
+ self.provider = provider
+
+ def count_for_identity(self, identity_hash: str) -> int:
+ row = self.provider.fetchone(
+ "SELECT COUNT(*) AS c FROM user_stickers WHERE identity_hash = ?",
+ (identity_hash,),
+ )
+ return int(row["c"]) if row else 0
+
+ def list_for_identity(self, identity_hash: str):
+ return self.provider.fetchall(
+ """
+ SELECT id, identity_hash, name, image_type, length(image_blob) AS image_size,
+ content_hash, source_message_hash, created_at, updated_at
+ FROM user_stickers
+ WHERE identity_hash = ?
+ ORDER BY updated_at DESC, id DESC
+ """,
+ (identity_hash,),
+ )
+
+ def get_row(self, sticker_id: int, identity_hash: str):
+ return self.provider.fetchone(
+ """
+ SELECT id, identity_hash, name, image_type, image_blob, content_hash,
+ source_message_hash, created_at, updated_at
+ FROM user_stickers
+ WHERE id = ? AND identity_hash = ?
+ """,
+ (sticker_id, identity_hash),
+ )
+
+ def delete(self, sticker_id: int, identity_hash: str) -> bool:
+ cur = self.provider.execute(
+ "DELETE FROM user_stickers WHERE id = ? AND identity_hash = ?",
+ (sticker_id, identity_hash),
+ )
+ return cur.rowcount > 0
+
+ def delete_all_for_identity(self, identity_hash: str) -> int:
+ cur = self.provider.execute(
+ "DELETE FROM user_stickers WHERE identity_hash = ?",
+ (identity_hash,),
+ )
+ return cur.rowcount
+
+ def update_name(
+ self, sticker_id: int, identity_hash: str, name: str | None
+ ) -> bool:
+ now = time.time()
+ cur = self.provider.execute(
+ """
+ UPDATE user_stickers
+ SET name = ?, updated_at = ?
+ WHERE id = ? AND identity_hash = ?
+ """,
+ (name, now, sticker_id, identity_hash),
+ )
+ return cur.rowcount > 0
+
+ def insert(
+ self,
+ identity_hash: str,
+ name: str | None,
+ image_type: str,
+ image_bytes: bytes,
+ source_message_hash: str | None = None,
+ ) -> dict | None:
+ """
+ Insert a sticker. Returns summary dict or None if duplicate (same content_hash).
+ """
+ if (
+ self.count_for_identity(identity_hash)
+ >= sticker_utils.MAX_STICKERS_PER_IDENTITY
+ ):
+ msg = "sticker_limit_reached"
+ raise ValueError(msg)
+
+ nt, ch = sticker_utils.validate_sticker_payload(image_bytes, image_type)
+ now = time.time()
+ try:
+ cur = self.provider.execute(
+ """
+ INSERT INTO user_stickers (
+ identity_hash, name, image_type, image_blob, content_hash,
+ source_message_hash, created_at, updated_at
+ )
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+ """,
+ (
+ identity_hash,
+ name,
+ nt,
+ image_bytes,
+ ch,
+ source_message_hash,
+ now,
+ now,
+ ),
+ )
+ except sqlite3.IntegrityError:
+ return None
+
+ new_id = cur.lastrowid
+ row = self.provider.fetchone(
+ """
+ SELECT id, identity_hash, name, image_type, length(image_blob) AS image_size,
+ content_hash, source_message_hash, created_at, updated_at
+ FROM user_stickers
+ WHERE id = ?
+ """,
+ (new_id,),
+ )
+ return dict(row) if row else None
+
+ def export_payloads_for_identity(self, identity_hash: str) -> list[dict]:
+ rows = self.provider.fetchall(
+ """
+ SELECT name, image_type, image_blob, source_message_hash
+ FROM user_stickers
+ WHERE identity_hash = ?
+ ORDER BY id ASC
+ """,
+ (identity_hash,),
+ )
+ out = []
+ for r in rows:
+ blob = r["image_blob"]
+ b64 = base64.b64encode(blob).decode("ascii")
+ out.append(
+ {
+ "name": r["name"],
+ "image_type": r["image_type"],
+ "image_bytes": b64,
+ "source_message_hash": r["source_message_hash"],
+ },
+ )
+ return out
+
+ def import_payloads(
+ self,
+ identity_hash: str,
+ items: list[dict],
+ *,
+ replace_duplicates: bool,
+ ) -> dict:
+ imported = 0
+ skipped_duplicates = 0
+ skipped_invalid = 0
+ errors: list[str] = []
+
+ for i, item in enumerate(items):
+ name = sticker_utils.sanitize_sticker_name(item.get("name"))
+ it = item.get("image_type")
+ b64 = item.get("image_bytes_b64")
+ src = item.get("source_message_hash")
+ try:
+ raw = base64.b64decode(b64, validate=False)
+ except (ValueError, TypeError):
+ skipped_invalid += 1
+ errors.append(f"decode_failed_at_{i}")
+ continue
+ try:
+ nt, ch = sticker_utils.validate_sticker_payload(raw, it)
+ except ValueError:
+ skipped_invalid += 1
+ errors.append(f"invalid_payload_at_{i}")
+ continue
+
+ existing = self.provider.fetchone(
+ "SELECT id FROM user_stickers WHERE identity_hash = ? AND content_hash = ?",
+ (identity_hash, ch),
+ )
+ if existing:
+ if not replace_duplicates:
+ skipped_duplicates += 1
+ continue
+ self.provider.execute(
+ "DELETE FROM user_stickers WHERE identity_hash = ? AND content_hash = ?",
+ (identity_hash, ch),
+ )
+
+ if (
+ self.count_for_identity(identity_hash)
+ >= sticker_utils.MAX_STICKERS_PER_IDENTITY
+ ):
+ errors.append("sticker_limit_reached")
+ break
+
+ now = time.time()
+ try:
+ self.provider.execute(
+ """
+ INSERT INTO user_stickers (
+ identity_hash, name, image_type, image_blob, content_hash,
+ source_message_hash, created_at, updated_at
+ )
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+ """,
+ (
+ identity_hash,
+ name,
+ nt,
+ raw,
+ ch,
+ src if isinstance(src, str) else None,
+ now,
+ now,
+ ),
+ )
+ imported += 1
+ except sqlite3.IntegrityError:
+ skipped_duplicates += 1
+
+ return {
+ "imported": imported,
+ "skipped_duplicates": skipped_duplicates,
+ "skipped_invalid": skipped_invalid,
+ "errors": errors,
+ }
diff --git a/meshchatx/src/backend/nomadnet_downloader.py b/meshchatx/src/backend/nomadnet_downloader.py
index d0528240..7f131c33 100644
--- a/meshchatx/src/backend/nomadnet_downloader.py
+++ b/meshchatx/src/backend/nomadnet_downloader.py
@@ -35,7 +35,8 @@ def sweep_stale_links():
"""Evict all non-ACTIVE links from the global cache."""
with _nomadnet_links_lock:
stale = [
- k for k, v in nomadnet_cached_links.items()
+ k
+ for k, v in nomadnet_cached_links.items()
if v.status is not RNS.Link.ACTIVE
]
for k in stale:
diff --git a/meshchatx/src/backend/sticker_utils.py b/meshchatx/src/backend/sticker_utils.py
new file mode 100644
index 00000000..0891507b
--- /dev/null
+++ b/meshchatx/src/backend/sticker_utils.py
@@ -0,0 +1,179 @@
+"""Validation and hashing for user sticker images stored per identity."""
+
+from __future__ import annotations
+
+import hashlib
+import base64
+
+MAX_STICKER_BYTES = 512 * 1024
+MAX_STICKERS_PER_IDENTITY = 2000
+
+_ALLOWED_TYPES = frozenset({"png", "jpeg", "jpg", "gif", "webp", "bmp"})
+
+_TYPE_ALIASES = {
+ "jpeg": "jpeg",
+ "jpg": "jpeg",
+ "pjpeg": "jpeg",
+ "png": "png",
+ "gif": "gif",
+ "webp": "webp",
+ "bmp": "bmp",
+}
+
+
+def normalize_image_type(image_type: str | None) -> str | None:
+ if not image_type:
+ return None
+ t = str(image_type).strip().lower()
+ t = t.removeprefix("image/")
+ t = _TYPE_ALIASES.get(t, t)
+ return t if t in _ALLOWED_TYPES else None
+
+
+def content_hash_hex(image_bytes: bytes) -> str:
+ return hashlib.sha256(image_bytes).hexdigest()
+
+
+def detect_image_format_from_magic(image_bytes: bytes) -> str | None:
+ """
+ Identify image format from file signature (magic bytes). Returns normalized
+ type key (png, jpeg, gif, webp, bmp) or None if unknown / too short / not allowed.
+ """
+ if not isinstance(image_bytes, (bytes, bytearray)) or len(image_bytes) < 4:
+ return None
+ b = bytes(image_bytes)
+ if len(b) >= 8 and b[:8] == b"\x89PNG\r\n\x1a\n":
+ return "png"
+ if len(b) >= 3 and b[0:3] == b"\xff\xd8\xff":
+ return "jpeg"
+ if len(b) >= 6 and b[0:6] in (b"GIF87a", b"GIF89a"):
+ return "gif"
+ if len(b) >= 12 and b[0:4] == b"RIFF" and b[8:12] == b"WEBP":
+ return "webp"
+ if len(b) >= 2 and b[0:2] == b"BM":
+ return "bmp"
+ return None
+
+
+def validate_sticker_payload(
+ image_bytes: bytes,
+ image_type: str | None,
+) -> tuple[str, str]:
+ """
+ Returns (normalized_image_type, content_hash_hex).
+
+ Declared image_type must match the format detected from magic bytes; stored
+ type is the normalized detected format.
+
+ Raises ValueError with a short reason on invalid input.
+ """
+ if not isinstance(image_bytes, (bytes, bytearray)):
+ msg = "invalid_image_bytes"
+ raise ValueError(msg)
+ if len(image_bytes) == 0:
+ msg = "empty_image"
+ raise ValueError(msg)
+ if len(image_bytes) > MAX_STICKER_BYTES:
+ msg = "image_too_large"
+ raise ValueError(msg)
+
+ nt = normalize_image_type(image_type)
+ if not nt:
+ msg = "invalid_image_type"
+ raise ValueError(msg)
+
+ detected = detect_image_format_from_magic(image_bytes)
+ if not detected:
+ msg = "invalid_image_signature"
+ raise ValueError(msg)
+ if detected != nt:
+ msg = "magic_type_mismatch"
+ raise ValueError(msg)
+
+ h = content_hash_hex(bytes(image_bytes))
+ return detected, h
+
+
+_EXPORT_FORMAT = "meshchatx-stickers"
+_EXPORT_VERSION = 1
+
+
+def validate_export_document(data: object) -> list[dict]:
+ """
+ Parse and lightly validate an import JSON document.
+ Returns a list of sticker dicts with keys: name, image_type, image_bytes (str base64),
+ source_message_hash (optional).
+ """
+ if not isinstance(data, dict):
+ msg = "invalid_document"
+ raise ValueError(msg)
+ if data.get("format") != _EXPORT_FORMAT:
+ msg = "invalid_format"
+ raise ValueError(msg)
+ try:
+ ver = int(data["version"])
+ except (KeyError, TypeError, ValueError) as exc:
+ msg = "unsupported_version"
+ raise ValueError(msg) from exc
+ if ver != _EXPORT_VERSION:
+ msg = "unsupported_version"
+ raise ValueError(msg)
+ stickers = data.get("stickers")
+ if not isinstance(stickers, list):
+ msg = "invalid_stickers_array"
+ raise ValueError(msg)
+ out: list[dict] = []
+ for i, item in enumerate(stickers):
+ if not isinstance(item, dict):
+ msg = f"invalid_sticker_at_{i}"
+ raise ValueError(msg)
+ name = item.get("name")
+ image_type = item.get("image_type")
+ image_b64 = item.get("image_bytes")
+ if not isinstance(image_b64, str) or not image_b64.strip():
+ msg = f"missing_image_bytes_at_{i}"
+ raise ValueError(msg)
+ try:
+ base64.b64decode(image_b64.strip(), validate=False)
+ except (ValueError, TypeError) as exc:
+ msg = f"invalid_base64_at_{i}"
+ raise ValueError(msg) from exc
+ src = item.get("source_message_hash")
+ out.append(
+ {
+ "name": name if isinstance(name, str) else None,
+ "image_type": image_type,
+ "image_bytes_b64": image_b64.strip(),
+ "source_message_hash": src if isinstance(src, str) else None,
+ },
+ )
+ return out
+
+
+def build_export_document(stickers: list[dict], exported_at_iso: str) -> dict:
+ """stickers: rows with name, image_type, image_bytes (base64 str), source_message_hash."""
+ return {
+ "format": _EXPORT_FORMAT,
+ "version": _EXPORT_VERSION,
+ "exported_at": exported_at_iso,
+ "stickers": stickers,
+ }
+
+
+def mime_for_image_type(normalized_type: str) -> str:
+ return {
+ "jpeg": "image/jpeg",
+ "png": "image/png",
+ "gif": "image/gif",
+ "webp": "image/webp",
+ "bmp": "image/bmp",
+ }.get(normalized_type, "application/octet-stream")
+
+
+def sanitize_sticker_name(name: str | None) -> str | None:
+ if name is None:
+ return None
+ s = "".join(ch for ch in str(name).strip() if ch.isprintable())
+ if not s:
+ return None
+ return s[:128]
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────